前言

之前我們測試過一些內建的Callback，這次筆者要自訂 Callback，將損失寫入 Pandas Data Frame，並且作圖畫出優化的軌跡。

可監看的事件

自訂 Callback 可監看的事件包括：

1. on_(train|test|predict)_begin：訓練、測試及預測開始時，可觸發事件。
2. on_(train|test|predict)_end：訓練、測試及預測結束時，可觸發事件。
3. on_(train|test|predict)_batch_begin：訓練、測試及預測每批開始時，可觸發事件。
4. on_(train|test|predict)_batch_end：訓練、測試及預測每批結束時，可觸發事件。
5. on_epoch_begin：每個執行週期開始時，可觸發事件。
6. on_epoch_end：每個執行週期結束時，可觸發事件。

以下是一個自訂的 Callback，定義以上所有事件的處理函數：

class CustomCallback(tf.keras.callbacks.Callback):
    def __init__(self):
        self.task_type=''
        self.epoch=0
        self.batch=0
        
    def on_train_begin(self, logs=None):
        self.task_type='訓練'
        print("訓練開始...")

    def on_train_end(self, logs=None):
        print("訓練結束.")

    def on_epoch_begin(self, epoch, logs=None):
        self.epoch=epoch
        print(f"{self.task_type}第 {epoch} 執行週期開始...")

    def on_epoch_end(self, epoch, logs=None):
        print(f"{self.task_type}第 {epoch} 執行週期結束.")

    def on_test_begin(self, logs=None):
        self.task_type='測試'
        print("測試開始...")

    def on_test_end(self, logs=None):
        print("測試結束.")

    def on_predict_begin(self, logs=None):
        self.task_type='預測'
        print("預測開始...")

    def on_predict_end(self, logs=None):
        print("預測結束.")

    def on_train_batch_begin(self, batch, logs=None):
        print(f"訓練 第 {self.epoch} 執行週期, 第 {batch} 批次開始...")

    def on_train_batch_end(self, batch, logs=None):
        print(f"訓練 第 {self.epoch} 執行週期, 第 {batch} 批次結束.")

    def on_test_batch_begin(self, batch, logs=None):
        print(f"測試 第 {self.epoch} 執行週期, 第 {batch} 批次開始...")

    def on_test_batch_end(self, batch, logs=None):
        print(f"測試 第 {self.epoch} 執行週期, 第 {batch} 批次結束.")

    def on_predict_batch_begin(self, batch, logs=None):
        print(f"預測 第 {self.epoch} 執行週期, 第 {batch} 批次開始...")

    def on_predict_batch_end(self, batch, logs=None):
        print(f"預測 第 {self.epoch} 執行週期, 第 {batch} 批次結束.")

在訓練、測試、預測使用此一 Callback 的程式碼如下：

# 訓練
model.fit(x_train_norm, y_train, epochs=5, batch_size=256, verbose=0, validation_split=0.2, callbacks=[CustomCallback()])

# 測試
model.evaluate(
    x_test_norm, y_test, batch_size=128, verbose=0, callbacks=[CustomCallback()]
)

# 預測
model.predict(x_test_norm, batch_size=128, callbacks=[CustomCallback()])

訓練顯示結果如下：

測試顯示結果如下：

範例檔案名稱為10_01_Custom_Callback.ipynb。

收集訓練過程的損失

我們可以透過自訂的 callback，在每一批的訓練結束都記錄損失，整個訓練結束後就可以畫出線圖如下。

class CustomCallback_2(tf.keras.callbacks.Callback):
    ...
    
    def on_train_batch_end(self, batch, logs=None):
        # 新增資料至 df2 DataFrame
        df2 = pd.DataFrame([[self.epoch, batch, logs["loss"]]], columns=['epoch', 'batch', 'metrics'])
        self.df = self.df.append(df2, ignore_index=True)

結果很有意思，優化的過程並不是損失函數一路減少，而是上上下下，但趨勢向下。

對執行週期(epoch)小計，取損失最小值，畫出線圖如下。

範例檔案名稱為10_02_Custom_Callback_loss.ipynb。

結論

透過自訂Callback，我們可以更深刻的理解優化的過程，除了損失函數，要取得其他資訊也是可行的，例如可以在事件處理函數中呼叫 self.model.get_weights() 取得每一神經層的所有權重。另外，自訂Callback，也可以用於除錯(Debug)，如果訓練出現錯誤，例如Nan或優化無法收斂，就可以逐批檢查相關變數。

本篇範例包括10_01_Custom_Callback.ipynb、10_02_Custom_Callback_loss.ipynb，可自【這裡】下載。